/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2023-2023. All rights reserved.
 * Description: asymmetric hierarchical concatenate 算法基类实现文件
 */

#include "asymmetric_hierarchical_concatenate_base_pub.h"

#include <iostream>
#include <fstream>

namespace hccl {

CommAHCBaseInfo::CommAHCBaseInfo(const std::vector<std::vector<u32>> &subGroups)
    : minSubGroupIdx_(0), maxSubGroupIdx_(0), subGroups_(subGroups)
{
    //rank 到 group index 的map 初始化以及最大最小分组下标的初始化
    u32 minSubGroupSize = subGroups_[0].size();
    u32 maxSubGroupSize = subGroups_[0].size();

    for (u32 i = 0; i < subGroups_.size(); ++i) {
        if (subGroups_[i].size() < minSubGroupSize) {
            minSubGroupSize = subGroups_[i].size();
            minSubGroupIdx_ = i;
        }
        if (subGroups_[i].size() > maxSubGroupSize) {
            maxSubGroupSize = subGroups_[i].size();
            maxSubGroupIdx_ = i;
        }
        for (u32 j = 0; j < subGroups_[i].size(); ++j) {
            rankGroupMap_.insert(std::make_pair(subGroups_[i][j], i));
        }
    }

    //当前仅支持静态配置两层拓扑的拼接算法类型，后续策略默认采用NB算法
    intraAlgType_ = ConcAlgType::CONC_ALG_TYPE_NB;
    if(subGroups_.size() <= AHC_SUBGROUP_THRESHOLD) {
        interAlgType_ = ConcAlgType::CONC_ALG_TYPE_RING;
    } else {
        interAlgType_ = ConcAlgType::CONC_ALG_TYPE_NB;
    }

    HCCL_DEBUG("[CommAHCBaseInfo] intraAlgType[%u] interAlgType[%u] minSubGroupSize[%u] maxSubGroupSize[%u]",
        static_cast<u32>(intraAlgType_), static_cast<u32>(interAlgType_), minSubGroupSize, maxSubGroupSize);
}

CommAHCBaseInfo::~CommAHCBaseInfo()
{
}

HcclResult CommAHCBaseInfo::Init()
{
    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::CheckSubGroups(std::vector<std::vector<u32>> &subGroups)
{
    if (subGroups.size() == 0) {
        HCCL_ERROR("[CommAHCBaseInfo] subGroups.size() == 0, subGroups init ERROR");
        return HCCL_E_PARA;
    }

    for (u32 i = 0; i < subGroups.size(); i++) {
        if (subGroups[i].size() == 0) {
            HCCL_ERROR("[CommAHCBaseInfo] subGroups[%u].size() == 0, subGroups[] init ERROR", i);
            return HCCL_E_PARA;
        }        
        for (u32 j = 0; j < subGroups[i].size(); j++){
            HCCL_DEBUG("[CommAHCBaseInfo] subGroups[%u][%u] = %u", i, j, subGroups[i][j]);
        }
    }
    return HCCL_SUCCESS;
}

const bool CommAHCBaseInfo::IsIntraAlgNB()
{
    return (intraAlgType_ == ConcAlgType::CONC_ALG_TYPE_NB);
}

const bool CommAHCBaseInfo::IsInterAlgNB()
{
    return (interAlgType_ == ConcAlgType::CONC_ALG_TYPE_NB);
}

HcclResult CommAHCBaseInfo::GetIntraCommGroup(u32 rank, std::vector<u32> &intraCommGroup)
{
    u32 groupIndex = rankGroupMap_[rank];
    intraCommGroup = subGroups_[groupIndex];

    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::GetInterCommGroupList(u32 rank, std::vector<std::vector<u32>> &interCommGroupList)
{
    //broke 方式的合法vetor大小为0或1,AHC 方式的合法vetor大小大于等于1
    for (u32 i = 0; i < logicCardCommGroups_.size(); ++i) {
        for (u32 j = 0; j < logicCardCommGroups_[i].size(); ++j) {
            if (rank == logicCardCommGroups_[i][j]) {
                interCommGroupList.push_back(logicCardCommGroups_[i]);
            }
        }
    }
    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::InputNBDstRanks(const u32 groupRank, const std::vector<u32> commGroups,
    std::set<u32> &dstRanks)
{
    for (auto i = 0; static_cast<u32>(1 << i) < commGroups.size(); ++i) {
        // 正方向第2^i个节点的rank号
        const u32 targetRankPos = static_cast<u32>(groupRank + (1 << i)) % commGroups.size();
        dstRanks.insert(commGroups[targetRankPos]);

        // 反方向第2^i个节点的rank号
        const u32 targetRankNeg = static_cast<u32>(groupRank + commGroups.size() - (1 << i)) % commGroups.size();
        dstRanks.insert(commGroups[targetRankNeg]);
    }

    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::InputRingDstRanks(const u32 groupRank, const std::vector<u32> commGroups,
    std::set<u32> &dstRanks)
{
    // 正方向下一个节点的rank号
    const u32 targetRankPos = static_cast<u32>(groupRank + 1) % commGroups.size();
    dstRanks.insert(commGroups[targetRankPos]);

    // 反方向上一个节点的rank号
    const u32 targetRankNeg = static_cast<u32>(groupRank + commGroups.size() - 1) % commGroups.size();
    dstRanks.insert(commGroups[targetRankNeg]);

    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::InitDstRanks(u32 rank, std::set<u32> &dstRanks)
{
    //组内和组间链接初始化
    std::vector<u32> intraCommGroup;
    std::vector<std::vector<u32>> interCommGroupList;

    GetIntraCommGroup(rank, intraCommGroup);
    GetInterCommGroupList(rank, interCommGroupList);

    if(intraAlgType_ == ConcAlgType::CONC_ALG_TYPE_RING) {
        InputRingDstRanks(GetIntraRank(rank), intraCommGroup, dstRanks);
    }
    else {
        InputNBDstRanks(GetIntraRank(rank), intraCommGroup, dstRanks);
    }

    if(interAlgType_ == ConcAlgType::CONC_ALG_TYPE_RING) {
        for (u32 i = 0; i < interCommGroupList.size(); ++i) {
            InputRingDstRanks(GetInterRank(rank), interCommGroupList[i], dstRanks);
        }
    }
    else {
        for (u32 i = 0; i < interCommGroupList.size(); ++i) {
            InputNBDstRanks(GetInterRank(rank), interCommGroupList[i], dstRanks);
        }
    }

    return HCCL_SUCCESS;
}

u32 CommAHCBaseInfo::GetIntraRank(const u32 rank)
{
    u32 intraRank = 0;
    for (u32 i = 0; i < subGroups_[rankGroupMap_[rank]].size(); i++) {
        if (subGroups_[rankGroupMap_[rank]][i] == rank) {
            intraRank = i;
            return intraRank;
        }
    }
    HCCL_DEBUG("[CommAHCBaseInfo][GetIntraRank] rank[%u] not found", rank);
    return intraRank;
}

u32 CommAHCBaseInfo::GetInterRank(const u32 rank)
{
    return rankGroupMap_[rank];
}

HcclResult CommAHCBaseInfo::CalcIntraSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count,
        const std::vector<LINK> &links, std::vector<LINK> &intraLinks, std::vector<Slice> &intraSlices)
{
    (void)rank;
    (void)dataUnitSize;
    (void)count;
    (void)links;
    (void)intraLinks;
    (void)intraSlices;
    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::CalcInterSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count,
        const std::vector<LINK> &links, std::vector<std::vector<LINK>> &interLinksVector,
        std::vector<std::vector<Slice>> &interSlicesVector, std::vector<u32> &logicCardList)
{
    (void)rank;
    (void)dataUnitSize;
    (void)count;
    (void)links;
    (void)interLinksVector;
    (void)interSlicesVector;
    (void)logicCardList;
    return HCCL_SUCCESS;
}

HcclResult CommAHCBaseInfo::GetExecutorOpInstance(const HcclCMDType opType, std::unique_ptr<AlgTemplateBase> &tempAlg,
const HcclDispatcher &dispatcher, const u64 reduceAttr)
{
    if(opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
        if (intraAlgType_ == ConcAlgType::CONC_ALG_TYPE_RING) {
            tempAlg.reset(new (std::nothrow) ReduceScatterRing(dispatcher, reduceAttr));
        }
        else {
            tempAlg.reset(new (std::nothrow) ReduceScatterNB(dispatcher, reduceAttr));
        }
    }
    else if(opType == HcclCMDType::HCCL_CMD_ALLREDUCE) {
        if (interAlgType_ == ConcAlgType::CONC_ALG_TYPE_RING) {
            tempAlg.reset(new (std::nothrow) AllReduceRing(dispatcher, reduceAttr));
        }
        else {
            tempAlg.reset(new (std::nothrow) AllReduceNB(dispatcher, reduceAttr));
        }
    }
    else if(opType == HcclCMDType::HCCL_CMD_ALLGATHER) {
        if (intraAlgType_ == ConcAlgType::CONC_ALG_TYPE_RING) {
            tempAlg.reset(new (std::nothrow) AllGatherRing(dispatcher));
        }
        else {
            tempAlg.reset(new (std::nothrow) AllGatherNB(dispatcher));
        }
    }
    CHK_SMART_PTR_NULL(tempAlg);
    return HCCL_SUCCESS;
}

bool CommAHCBaseInfo::IsNeedInterProc(const u32 rank)
{
    (void)rank;
    return true;
}

CommBrokeAlignInfo::CommBrokeAlignInfo(const std::vector<std::vector<u32>> &subGroups)
    : CommAHCBaseInfo(subGroups)
{
}

CommBrokeAlignInfo::~CommBrokeAlignInfo()
{
}

HcclResult CommBrokeAlignInfo::Init()
{
    // 参数检查
    CHK_RET(CheckSubGroups(subGroups_));

    //初始化broke 对齐的组间通信域相关信息
    for (u32 i = 0; i < subGroups_[minSubGroupIdx_].size(); ++i) {
        std::vector<u32> logicGroup;
        for (u32 j = 0; j < subGroups_.size(); ++j) {
            logicGroup.push_back(subGroups_[j][i]);
        }
        logicCardCommGroups_.push_back(logicGroup);
    }
    return HCCL_SUCCESS;
}

bool CommBrokeAlignInfo::IsNeedInterProc(const u32 rank)
{
    u32 intraRank = GetIntraRank(rank);
    HCCL_DEBUG("[CommBrokeAlignInfo][IsNeedInterProc] rank[%u] intraRank[%u] minSize[%u]",
        rank, intraRank, subGroups_[minSubGroupIdx_].size());
    if ( intraRank > (subGroups_[minSubGroupIdx_].size() - 1)) {
        return false;
    }
    return true;
}

HcclResult CommBrokeAlignInfo::CalcIntraSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count,
        const std::vector<LINK> &links, std::vector<LINK> &intraLinks, std::vector<Slice> &intraSlices)
{
    // 计算组内每个rank结果上的offset和size
    HCCL_DEBUG("[CommBrokeAlignInfo][CalcIntraSlicesAndLinks] begin calc intra slices and links rank[%u]", rank);

    u64 sliceSizeCalculated = (count + (static_cast<u32>(subGroups_[minSubGroupIdx_].size()) - 1)) / subGroups_[minSubGroupIdx_].size() * dataUnitSize;
    u64 totalSize = count * dataUnitSize;
    u64 residueSize = totalSize;
    u64 sliceSizeAligned;
    const u64 sizeAlignedMinSize = 128 * 1024; // 优化小包性能，小于128k不切片
    if (sliceSizeCalculated > sizeAlignedMinSize) {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
    } else {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, sizeAlignedMinSize);
    }

    for (u32 i = 0; i < subGroups_[rankGroupMap_[rank]].size(); ++i) {
        intraLinks.push_back(links[subGroups_[rankGroupMap_[rank]][i]]);
        Slice slice;
        if (i < subGroups_[minSubGroupIdx_].size()) {
            slice.size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
            slice.offset = totalSize - residueSize;
            residueSize -= slice.size;
        } else {
            slice.size = 0;
            slice.offset = totalSize - residueSize;
        }
        HCCL_DEBUG("[CommBrokeAlignInfo][CalcIntraSlicesAndLinks] rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu",
            rank, i, slice.offset, i, slice.size);
        intraSlices.push_back(slice);
    }

    HCCL_DEBUG("[CommBrokeAlignInfo][CalcIntraSlicesAndLinks] end calc intra slices and links rank[%u]", rank);

    return HCCL_SUCCESS;
}

HcclResult CommBrokeAlignInfo::CalcInterSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count, const std::vector<LINK> &links,
        std::vector<std::vector<LINK>> &interLinksVector, std::vector<std::vector<Slice>> &interSlicesVector, std::vector<u32> &logicCardList)
{
    (void)logicCardList;
    // 查找自己位于组内的第几个rank
    HCCL_DEBUG("[CommBrokeAlignInfo][CalcInterSlicesAndLinks] begin calc inter slices and links rank[%u]", rank);

    u32 intraRank = GetIntraRank(rank);

    std::vector<Slice> intraSlices;
    std::vector<LINK> intraLinks;
    CHK_RET(CalcIntraSlicesAndLinks(rank, dataUnitSize, count, links, intraLinks, intraSlices));

    // 计算组间每个rank结果上的offset和size
    u64 sliceSizeCalculated =
        (intraSlices[intraRank].size / dataUnitSize + (static_cast<u32>(subGroups_.size()) - 1)) / subGroups_.size() * dataUnitSize;
    u64 totalSize = intraSlices[intraRank].size;
    u64 residueSize = totalSize;
    u64 sliceSizeAligned;
    const u64 sizeAlignedMinSize = 128 * 1024; // 优化小包性能，小于128k不切片
    if (sliceSizeCalculated > sizeAlignedMinSize) {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
    } else {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, sizeAlignedMinSize);
    }

    std::vector<LINK>  interLinks;
    std::vector<Slice> interSlices;
    for (u32 i = 0; i < subGroups_.size(); ++i) {
        interLinks.push_back(links[subGroups_[i][intraRank]]);
        Slice slice;
        slice.size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
        slice.offset = intraSlices[intraRank].offset + totalSize - residueSize;
        residueSize -= slice.size;
        HCCL_DEBUG("[CommBrokeAlignInfo][CalcInterSlicesAndLinks] rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu",
            rank, i, slice.offset, i, slice.size);
        interSlices.push_back(slice);
    }
    interLinksVector.push_back(interLinks);
    interSlicesVector.push_back(interSlices);

    HCCL_DEBUG("[CommBrokeAlignInfo][CalcInterSlicesAndLinks] end calc inter slices and links rank[%u]", rank);
    return HCCL_SUCCESS;
}

CommAHCAlignInfo::CommAHCAlignInfo(const std::vector<std::vector<u32>> &subGroups)
    : CommAHCBaseInfo(subGroups)
{
}

CommAHCAlignInfo::~CommAHCAlignInfo()
{
}

HcclResult CommAHCAlignInfo::Init()
{
    // 参数检查
    CHK_RET(CheckSubGroups(subGroups_));

    //初始化slice相关信息
    CHK_RET(InitSliceInfo());

    //计算 logicCard 相关信息;
    InitLogicCardInfo();

    //初始化相关Map信息
    CHK_RET(InitMapInfo());

    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::InitSliceInfo()
{
    // 计算 totalSliceSegment_ ，即所有分组大小的最小公倍数
    totalSliceSegment_ = subGroups_[0].size();
    //u32 groupSizeGcd;
    for (u32 i = 1; i < subGroups_.size(); ++i) {
        u32 groupSize = static_cast<u32>(subGroups_[i].size());
        totalSliceSegment_ = totalSliceSegment_ * groupSize / std::__gcd(totalSliceSegment_, groupSize);
    }
     HCCL_DEBUG("[CommAHCAlignInfo][InitSliceInfo] totalSliceSegment [%u]", totalSliceSegment_);

    //计算 logicCardSliceSize_ ;
    std::set<u32> sliceOffset;
    for (u32 i = 0; i < subGroups_.size(); ++i) {
        for (u32 j = 0; j < subGroups_[i].size(); ++j) {
            u32 rankSliceSize = (totalSliceSegment_ / subGroups_[i].size()) * (j + 1);
            sliceOffset.insert(rankSliceSize);
            HCCL_DEBUG("[CommAHCAlignInfo][InitSliceInfo] sliceOffset [%u]",rankSliceSize);
        }
    }
    sliceOffset.insert(static_cast<u32>(0));
    logicCardSliceOffset_.resize(sliceOffset.size());
    std::copy(sliceOffset.begin(), sliceOffset.end(), logicCardSliceOffset_.begin());

    std::vector<u32>::iterator itPre = logicCardSliceOffset_.begin();
    std::vector<u32>::iterator itNext = logicCardSliceOffset_.begin();
    itNext++;
    while(itNext != logicCardSliceOffset_.end()) {
        auto boundDiff = (*itNext) - (*itPre);
        logicCardSliceSize_.push_back(boundDiff);
        itPre++;
        itNext++;
    }

    CHK_PRT_RET(logicCardSliceSize_.size() !=(logicCardSliceOffset_.size() - 1),
        HCCL_ERROR("[CommAHCAlignInfo][InitSliceInfo] cardOffset size [%u]  cardSize size [%u] check error", 
        logicCardSliceSize_.size(),logicCardSliceOffset_.size() ), HCCL_E_INTERNAL);

    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::InitLogicCardInfo()
{
    //计算 logicCardCommGroups_;
    for (std::vector<u32>::iterator it = (logicCardSliceOffset_.begin() + 1); it != logicCardSliceOffset_.end(); ++it) {
        std::vector<u32> logicGroup;
        for (u32 i = 0; i < subGroups_.size(); ++i) {
            u32 logicRank;
            if ((*it) % (totalSliceSegment_ / subGroups_[i].size()) != 0) {
                logicRank = (*it) / (totalSliceSegment_ / subGroups_[i].size()) + 1;
            } else {
                logicRank = (*it) / (totalSliceSegment_ / subGroups_[i].size());
            }
            logicGroup.push_back(subGroups_[i][logicRank - 1]);
        }
        logicCardCommGroups_.push_back(logicGroup);
    }

    //计算 logicCardGroup_
    u32 curRank = subGroups_[minSubGroupIdx_][0];
    u32 curOffset = 0;
    std::vector<u32>::iterator it = logicCardSliceOffset_.begin();
    u32 curIdx = 0;
    u32 curLogicIdx = 0;
    logicCardGroup_.resize(static_cast<u32>(subGroups_[minSubGroupIdx_].size()));
    for (u32 i = 0; i < logicCardCommGroups_.size(); ++i) {
        if (logicCardCommGroups_[i][minSubGroupIdx_] != curRank) {
            logicCardGroup_[curLogicIdx].resize(i - curIdx);
            for (u32 j = 0; j < i - curIdx; j++) {
                logicCardGroup_[curLogicIdx][j] = curIdx + j;
            }
            curIdx = i;
            curLogicIdx++;
            curRank = logicCardCommGroups_[i][minSubGroupIdx_];
            curOffset = *it;
        }
        logicCardExecuteOffset_.push_back(*it - curOffset);
        it++;
    }
    if (curIdx != logicCardCommGroups_.size() - 1) {
        logicCardGroup_[curLogicIdx].resize(logicCardCommGroups_.size() - curIdx);
        for (u32 i = 0; i < logicCardCommGroups_.size() - curIdx; i++) {
            logicCardGroup_[curLogicIdx][i] = curIdx + i;
        }
    }
    return HCCL_SUCCESS;
}

bool CommAHCAlignInfo::CompareLogicCardExcuteOrder(u32 i, u32 j)
{
    return logicCardExecuteOffset_[i] < logicCardExecuteOffset_[j];
}

HcclResult CommAHCAlignInfo::InitMapInfo()
{
    //rank 到 logicCardOrder  初始化
    for (u32 i = 0; i < logicCardCommGroups_.size(); ++i) {
        for (u32 j = 0; j < logicCardCommGroups_[i].size(); ++j) {
            rankLogicCardOrderMap_[logicCardCommGroups_[i][j]].push_back(i);
            rankLogicCardMap_[logicCardCommGroups_[i][j]].push_back(i);
        }
    }

    //定义 lambda 将对象指针传递到成员函数
    auto  sortLambda = [this](u32 i, u32 j) {
        return this->CompareLogicCardExcuteOrder(i,j);
    };

    //rankLogicCardOrderMap_ 内的逻辑同号卡list按照 logicCardExecuteOffset_ 并发流开始时间排序
    for (auto iter = rankLogicCardOrderMap_.begin(); iter != rankLogicCardOrderMap_.end(); iter++) {
        std::vector<u32> &rankLogicCardList = iter->second;
        std::sort(rankLogicCardList.begin(), rankLogicCardList.end(), sortLambda);
    }

    return HCCL_SUCCESS;
}

//获取当前rank对应的多个逻辑同号卡,并且按照并发流的开始执行时间排序
HcclResult CommAHCAlignInfo::GetLogicCardExecuteOrder(u32 rank, std::vector<u32> &executeOrder)
{
    executeOrder = rankLogicCardOrderMap_[rank];
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::CalcIntraSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count,
        const std::vector<LINK> &links, std::vector<LINK> &intraLinks, std::vector<Slice> &intraSlices)
{
    // 计算组内每个rank结果上的offset和size
    HCCL_DEBUG("[CommAHCAlignInfo][CalcIntraSlicesAndLinks] begin calc intra slices and links rank[%u]", rank);

    u64 sliceSizeCalculated = (count + (totalSliceSegment_ * static_cast<u32>(subGroups_.size()) - 1))
        / (totalSliceSegment_ * subGroups_.size()) * dataUnitSize;
    u64 totalSize = count * dataUnitSize;
    u64 residueSize = totalSize;
    u64 sliceSizeAligned;
    const u64 sizeAlignedMinSize = 128 * 1024; // 优化小包性能，小于128k不切片
    if (sliceSizeCalculated > sizeAlignedMinSize) {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
    } else {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, sizeAlignedMinSize);
    }
    sliceSizeAligned = sliceSizeAligned * static_cast<u32>(subGroups_.size()) * 
        (totalSliceSegment_ / static_cast<u32>(subGroups_[rankGroupMap_[rank]].size()));

    for (u32 i = 0; i < subGroups_[rankGroupMap_[rank]].size(); ++i) {
        intraLinks.push_back(links[subGroups_[rankGroupMap_[rank]][i]]);
        Slice slice;
        slice.size = (residueSize > sliceSizeAligned) ? sliceSizeAligned : residueSize;
        slice.offset = totalSize - residueSize;
        residueSize -= slice.size;
        HCCL_DEBUG("[CommAHCAlignInfo][CalcIntraSlicesAndLinks] rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu",
            rank, i, slice.offset, i, slice.size);
        intraSlices.push_back(slice);
    }

    HCCL_DEBUG("[CommAHCAlignInfo][CalcIntraSlicesAndLinks] end calc intra slices and links rank[%u]", rank);
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::PrepareWholeLogicSlices(const Slice intraSlice, const u64 sliceSizeAligned, const u32 originOffset,
        std::vector<Slice> &logicGroupSlice, std::vector<u32> &logicCardList)
{
    for (u32 i = 0; i < logicCardList.size(); i++) {
        Slice logicSlice;
        u32 logicRank = logicCardList[i];
        // 计算当前逻辑同号组的offset大小
        u32 offsetDiff = logicCardSliceOffset_[logicRank + 1] - logicCardSliceOffset_[logicRank];
        HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] logicGroupSlice begin, logicRank : [%u]," \
            "offsetDiff : [%u], offset_next : [%u], offset_cur[%u]", logicRank, offsetDiff,
            logicCardSliceOffset_[logicRank + 1], logicCardSliceOffset_[logicRank]);

        logicSlice.size = sliceSizeAligned * offsetDiff;
        logicSlice.offset = intraSlice.offset + sliceSizeAligned * (logicCardSliceOffset_[logicRank] - originOffset);
        HCCL_DEBUG("[CommAHCAlignInfo][PrepareFullLogicSlices] logicGroupSlice end, logicRank : [%u] ," \
            "size : [%u], offset : [%u] ", logicRank, logicSlice.size, logicSlice.offset);
        logicGroupSlice.push_back(logicSlice);
    }
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::PreparePartialLogicSlices(const Slice intraSlice, const u64 sliceSizeAligned, const u32 originOffset,
        std::vector<Slice> &logicGroupSlice, std::vector<u32> &logicCardList)
{
    for (u32 i = 0; i < logicCardList.size(); i++) {
        Slice logicSlice;
        u32 logicRank = logicCardList[i];
        // 计算当前逻辑同号组的offset大小
        u32 offsetDiff = logicCardSliceOffset_[logicRank + 1] - logicCardSliceOffset_[logicRank];
        HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] logicGroupSlice begin, logicRank : [%u]," \
            "offsetDiff : [%u], offset_next : [%u], offset_cur[%u]", logicRank, offsetDiff, 
            logicCardSliceOffset_[logicRank + 1], logicCardSliceOffset_[logicRank]);

        // 当前rank在组内对应的offset能获取到完全的数据，即前几个逻辑同号卡
        if ((logicCardSliceOffset_[logicRank + 1] - originOffset) <= intraSlice.size / sliceSizeAligned) { 
            logicSlice.size = sliceSizeAligned * offsetDiff;
            logicSlice.offset = intraSlice.offset + sliceSizeAligned * (logicCardSliceOffset_[logicRank] - originOffset);
        // 当前rank在组内对应的offset能获取到部分的数据，即边界上的逻辑同号卡
        } else if ((logicCardSliceOffset_[logicRank] - originOffset) <= intraSlice.size / sliceSizeAligned){ 
            logicSlice.size = intraSlice.size - (logicCardSliceOffset_[logicRank] - originOffset) * sliceSizeAligned;
            logicSlice.offset = intraSlice.offset + sliceSizeAligned * (logicCardSliceOffset_[logicRank] - originOffset);
        // 当前rank在组内对应的offset不能获取到数据，即最后的逻辑同号卡
        } else {
            logicSlice.size = 0;
            logicSlice.offset = 0;
        }
        HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] logicGroupSlice end, logicRank : [%u] ," \
            "size : [%u], offset : [%u] ",logicRank, logicSlice.size, logicSlice.offset);

        logicGroupSlice.push_back(logicSlice);
    }
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::PrepareEmptyLogicSlices(std::vector<Slice> &logicGroupSlice,
    const std::vector<u32> &logicCardList)
{
    for (u32 i = 0; i < logicCardList.size(); i++) {
        Slice logicSlice;
        logicSlice.size = 0;
        logicSlice.offset = 0;
        logicGroupSlice.push_back(logicSlice);
    }
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::CalcLogicSlicesAndLinks(std::vector<Slice> &logicGroupSlice, std::vector<u32> &logicCardList,
        const std::vector<LINK> &links, std::vector<std::vector<LINK>> &interLinksVector,
        std::vector<std::vector<Slice>> &interSlicesVector)
{
    for (u32 i = 0; i < logicGroupSlice.size(); i++) {
        Slice logicSlice = logicGroupSlice[i];
        std::vector<Slice> interSlices;
        std::vector<LINK> interLinks;
        u32 logicRank = logicCardList[i];
        u64 totalSize = logicSlice.size;
        u64 residueSize = totalSize;
        u64 logicSliceSizeAligned;
        if (logicSlice.size % subGroups_.size() == 0 && logicSlice.size % HCCL_MIN_SLICE_ALIGN == 0) {
            logicSliceSizeAligned = logicSlice.size / subGroups_.size();
        } else {
            u64 sliceSizeCalculated = (logicSlice.size + static_cast<u32>(subGroups_.size()) - 1) / static_cast<u32>(subGroups_.size());
            logicSliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
        }
        for (u32 j = 0; j < subGroups_.size(); j++) {
            u32 curRank = logicCardCommGroups_[logicRank][j];
            Slice slice;
            interLinks.push_back(links[curRank]);
            slice.size = (residueSize > logicSliceSizeAligned) ? logicSliceSizeAligned : residueSize;
            slice.offset = logicSlice.offset + totalSize - residueSize;
            residueSize -= slice.size;
            interSlices.push_back(slice);
        }
        interLinksVector.push_back(interLinks);
        interSlicesVector.push_back(interSlices);
    }
    return HCCL_SUCCESS;
}

HcclResult CommAHCAlignInfo::CalcInterSlicesAndLinks(const u32 rank, const u32 dataUnitSize, const u64 count,
        const std::vector<LINK> &links, std::vector<std::vector<LINK>> &interLinksVector,
        std::vector<std::vector<Slice>> &interSlicesVector, std::vector<u32> &logicCardList)
{
    HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] begin calc inter slices and links rank[%u]", rank);

    u32 intraRank = GetIntraRank(rank);

    std::vector<Slice> intraSlices;
    std::vector<LINK> intraLinks;
    
    CHK_RET(CalcIntraSlicesAndLinks(rank, dataUnitSize, count, links, intraLinks, intraSlices));
    GetLogicCardExecuteOrder(rank, logicCardList);

    // 计算当前rank逻辑同号卡之间最小slice的大小
    u64 sliceSizeCalculated = (count + (totalSliceSegment_ * static_cast<u32>(subGroups_.size()) - 1))
        / (totalSliceSegment_ * subGroups_.size()) * dataUnitSize;
    const u64 sizeAlignedMinSize = 128 * 1024; // 优化小包性能，小于128k不切片
    u64 sliceSizeAligned;
    if (sliceSizeCalculated > sizeAlignedMinSize) {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, HCCL_MIN_SLICE_ALIGN);
    } else {
        sliceSizeAligned = AlgTemplateBase::RoundUpWithDivisor(sliceSizeCalculated, sizeAlignedMinSize);
    }

    sliceSizeAligned = sliceSizeAligned * static_cast<u32>(subGroups_.size());
    u32 originOffset = intraRank *  totalSliceSegment_ / subGroups_[rankGroupMap_[rank]].size(); // 当前rank起始offset
    
    HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] rank : [%u], intraslice.size : [%u], intraslice.offset : [%u]," \
        "sliceSizeAligned : [%u], originOffset : [%u]", rank, intraSlices[intraRank].size, intraSlices[intraRank].offset,
        sliceSizeAligned, originOffset);

    // 进行逻辑同号组对应的slice切分
    std::vector<Slice> logicGroupSlice; // 逻辑同号组对应的slice
    HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] check rank : [%u], intraSlices[%u].size : [%u], sliceSizeAligned : [%u]," \
        "totalSliceSegment_ : [%u], subGroups_[rankGroupMap_[rank]].size : [%u]", rank, intraRank, intraSlices[intraRank].size, 
        sliceSizeAligned, totalSliceSegment_, subGroups_[rankGroupMap_[rank]].size());

    // 当前rank有完整的对齐后的数据量
    if (intraSlices[intraRank].size / sliceSizeAligned == totalSliceSegment_ / subGroups_[rankGroupMap_[rank]].size()) {
        CHK_RET(PrepareWholeLogicSlices(intraSlices[intraRank], sliceSizeAligned, originOffset, logicGroupSlice, logicCardList));
    // 当前rank有不完整的数据量
    } else if (intraSlices[intraRank].size != 0) {
        CHK_RET(PreparePartialLogicSlices(intraSlices[intraRank], sliceSizeAligned, originOffset, logicGroupSlice, logicCardList));
    } else {
        CHK_RET(PrepareEmptyLogicSlices(logicGroupSlice, logicCardList));
    }

    // 计算当前rank逻辑同号组之间的slice大小
    CHK_RET(CalcLogicSlicesAndLinks(logicGroupSlice, logicCardList, links, interLinksVector, interSlicesVector));

    HCCL_DEBUG("[CommAHCAlignInfo][CalcInterSlicesAndLinks] end calc inter slices and links rank[%u]", rank);
    return HCCL_SUCCESS;
}


AHCExecutorBase::AHCExecutorBase(const HcclDispatcher dispatcher, const u64 reduceAttrBitMap,
    const u64 totalCount, const std::vector<std::vector<u32>> &subGroups)
    : AlgTemplateBase(dispatcher), rankSize_(1), reduceAttr_(reduceAttrBitMap), subGroups_(subGroups), fftsPhase_(1),
    totalCount_(totalCount)
{
}

AHCExecutorBase::~AHCExecutorBase()
{
}

HcclResult AHCExecutorBase::CommAHCInfoInit(std::vector<std::vector<u32>> &subGroups)
{
    (void) subGroups;
    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::RunAsync(const u32 rank, const u32 rankSize,
    const std::vector<LINK> &links)
{  
    HCCL_INFO("[AHCExecutorBase][RunAsync] start rank[%u] rankSize[%u]", rank, rankSize);

    HcclResult ret = HCCL_SUCCESS;
    ret = PrepareRunAsync(rank, rankSize, links);

    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in PrepareRunAsync step", rank, count_), ret);

    CHK_PRT_RET(rankSize == 1, HCCL_INFO("[AHCExecutorBase][RunAsync] rankSize[%u], do nothing.",
        rankSize), HCCL_SUCCESS);

    HCCL_DEBUG("[AHCExecutorBase][RunAsync] rank[%u] begin intra rs", rank);

    ret = RunIntraReduceScatter(rank, links, commAHCBaseInfo_);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
        "RunIntraReduceScatter  step", rank, count_), ret);

    HCCL_DEBUG("[AHCExecutorBase][RunAsync] rank[%u] end intra rs begin inter", rank);

    // 垂直方向做allreduce ring
    ret = RunInterAllReduce(rank, links, commAHCBaseInfo_);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
        "RunInterAllReduce step", rank, count_), ret);

    HCCL_DEBUG("[AHCExecutorBase][RunAsync] rank[%u] end inter begin intra ag", rank);

    // 水平方向做broken allgather ring
    ret = RunIntraAllGather(rank, links, commAHCBaseInfo_);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
        "RunIntraAllGather step", rank, count_), ret);

    HCCL_DEBUG("[AHCExecutorBase][RunAsync] rank[%u] end intra ag", rank);

    HCCL_INFO("[AHCExecutorBase][RunAsync] finished: rank[%u]", rank);
    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::RunAsyncStaged(const u32 rank, const u32 rankSize, const std::vector<LINK> &links,
    RunStage stage)
{
    CHK_PRT_RET(rankSize == 1 && stage != RunStage::RUN_PREPARE,
        HCCL_INFO("[AHCExecutorBase][RunAsyncStaged] rankSize[%u], stage[%d], do nothing.",
        rankSize, stage), HCCL_SUCCESS);

    HCCL_INFO("AHCExecutorBase RunAsyncStaged start: rank[%u]", rank);

    HcclResult ret = HCCL_SUCCESS;
    switch (stage) {
        case RunStage::RUN_PREPARE:
            ret = PrepareRunAsync(rank, rankSize, links);
            CHK_PRT_RET(ret != HCCL_SUCCESS,
                HCCL_ERROR("[AHCExecutorBase][RunAsyncStaged]rank[%u] count[%llu] failed in PrepareRunAsync step",
                rank, count_), ret);
            break;
        case RunStage::RUN_REDUCE_SCATTER:
            // 水平方向做broken reducescatter ring
            ret = RunIntraReduceScatter(rank, links, commAHCBaseInfo_);
            CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
                "RunIntraReduceScatter  step", rank, count_), ret);
            break;
        case RunStage::RUN_ALLREDUCE:
            // 垂直方向做allreduce ring
            ret = RunInterAllReduce(rank, links, commAHCBaseInfo_);
            CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
                "RunInterAllReduce step", rank, count_), ret);
            break;
        case RunStage::RUN_ALLGATHER:
            // 水平方向做broken allgather ring
            ret = RunIntraAllGather(rank, links, commAHCBaseInfo_);
            CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AHCExecutorBase][RunAsync]rank[%u] count[%llu] failed in "\
                "RunIntraAllGather step", rank, count_), ret);
            break;
        default:
            HCCL_ERROR("[AHCExecutorBase][RunAsyncStaged]stage[%d]is not support", stage);
            return HCCL_E_NOT_SUPPORT;
    }
    HCCL_INFO("AHCExecutorBase RunAsyncStaged stage[%d] finished: rank[%u] ranksize[%u]", stage, rank, rankSize);
    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::PrepareRunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
{
    HcclResult ret = HCCL_SUCCESS;
    CHK_SMART_PTR_NULL(dispatcher_);
    CHK_PTR_NULL(stream_.ptr());
    CHK_PRT_RET(!outputMem_ || !inputMem_,
        HCCL_ERROR("[AHCExecutorBase][PrepareRunAsync]rank[%u] run_async inputmem or outputmem is null", rank), HCCL_E_PTR);

    HCCL_INFO("AHCExecutorBase run: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", \
              rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);

    CHK_PRT_RET(links.size() < rankSize, HCCL_ERROR("[AHCExecutorBase][PrepareRunAsync]rank[%u] linksize[%llu] is less "\
        "than rankSize[%u]", rank, links.size(), rankSize), HCCL_E_INTERNAL);

    // 如果ranksize为1, inline reduce和普通跨片reduce操作一致，从input->output
    if (rankSize == 1) {
        if (inputMem_ != outputMem_) {
            ret = HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_);
            CHK_PRT_RET(ret != HCCL_SUCCESS,
                HCCL_ERROR("[AHCExecutorBase][PrepareRunAsync]rank[%u] memcpy async failed", rank), ret);
        }
        return ret;
    }

    rankSize_ = rankSize;

    CHK_RET(CommAHCInfoInit(subGroups_));

    // 检查、并清空slices_
    if (slices_.size() != 0) {
        HCCL_WARNING("[AHCExecutorBase][PrepareRunAsync]AHCExecutorBase not supported passing in parameter slice_, "\
            "otherwise will be cleared");
        slices_.clear();
    }

    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::RunIntraReduceScatter(const u32 rank, const std::vector<LINK> &links,
    const std::unique_ptr<CommAHCBaseInfo> &commAHCBaseInfo)
{
    // 获取当前rank的组内rank
    HCCL_INFO("[AHCExecutorBase][RunIntraReduceScatter] begin intra reduce scatter rank[%u]", rank);

    u32 intraRank = commAHCBaseInfo->GetIntraRank(rank);

    // 创建执行算子实列
    std::unique_ptr<AlgTemplateBase> tempAlg;
    CHK_RET(commAHCBaseInfo->GetExecutorOpInstance(HcclCMDType::HCCL_CMD_REDUCE_SCATTER, tempAlg, dispatcher_, reduceAttr_));

    std::vector<Slice> intraSlices;
    std::vector<LINK> intraLinks;
    CHK_RET(commAHCBaseInfo->CalcIntraSlicesAndLinks(rank, DataUnitSize(dataType_), count_, links, intraLinks, intraSlices));

    // 长度不足2，直接跳过
    if (intraLinks.size() < 2) {
        return HCCL_SUCCESS;
    }

    HCCL_DEBUG("[AHCExecutorBase][RunIntraReduceScatter] run inst rank[%u] intraRank[%u], IntraSize=%u",
        rank, intraRank, intraLinks.size());

    CHK_RET(RunInstance(intraRank, intraLinks, intraSlices, tempAlg, HcclCMDType::HCCL_CMD_REDUCE_SCATTER));

    HCCL_DEBUG("[AHCExecutorBase][RunIntraReduceScatter] end intra reduce scatter rank[%u]", rank);

    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::RunInterAllReduce(const u32 rank, const std::vector<LINK> &links, 
    const std::unique_ptr<CommAHCBaseInfo> &commAHCBaseInfo)
{
    (void) rank;
    (void) links;
    (void) commAHCBaseInfo;

    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::RunIntraAllGather(const u32 rank, const std::vector<LINK> &links,
    const std::unique_ptr<CommAHCBaseInfo> &commAHCBaseInfo)
{
    HCCL_INFO("[AHCExecutorBase][RunIntraAllGather] begin intra allgather rank[%u]", rank);

    // 获取当前rank的组内rank
    u32 intraRank = commAHCBaseInfo->GetIntraRank(rank);

    // 创建执行算子实列
    std::unique_ptr<AlgTemplateBase> tempAlg;
    CHK_RET(commAHCBaseInfo->GetExecutorOpInstance(HcclCMDType::HCCL_CMD_ALLGATHER, tempAlg, dispatcher_, reduceAttr_));

    std::vector<Slice> intraSlices;
    std::vector<LINK> intraLinks;

    CHK_RET(commAHCBaseInfo->CalcIntraSlicesAndLinks(rank, DataUnitSize(dataType_), count_, links, intraLinks, intraSlices));

    // 长度不足2，直接跳过
    if (intraLinks.size() < 2) {
        return HCCL_SUCCESS;
    }

    HCCL_DEBUG("[AHCExecutorBase][RunIntraAllGather] run inst rank[%u] intraRank[%u], IntraSize=%u",
        rank, intraRank, intraLinks.size());

    CHK_RET(RunInstance(intraRank, intraLinks, intraSlices, tempAlg, HcclCMDType::HCCL_CMD_ALLGATHER));

    HCCL_DEBUG("[AHCExecutorBase][RunIntraAllGather] end intra allgather rank[%u]", rank);
    return HCCL_SUCCESS;
}

HcclResult AHCExecutorBase::FftsRsetPhase(const std::vector<Slice> &slices)
{
    HCCL_DEBUG("AHCExecutorBase phase[%u] begin", fftsPhase_);

    bool isFftsCtxInitialized = true;  // 截断子图之前要确保前面子图存在
    CHK_RET(IsCtxInitialized(dispatcher_, &isFftsCtxInitialized));
    if (isFftsCtxInitialized) {
        CHK_RET(LaunchTask(dispatcher_, stream_));
    }

    u64 sliceNum = 0;
    for (u32 i = 0; i < slices.size(); i++) {
        if (slices[i].size > 0) {
            sliceNum++;
        }
    }
    sliceNum = (sliceNum | static_cast<u64>(fftsPhase_) << 32);//复用高32位区分phase

    HCCL_DEBUG("AHCExecutorBase phase[%u] FFTS key[%llu]", fftsPhase_, sliceNum);

    const u64 totalSize = totalCount_ * SIZE_TABLE[dataType_];
    bool hugeData = totalSize / rankSize_  > RDMA_SEND_MAX_SIZE ||
                    totalSize > SDMA_SEND_MAX_SIZE;

    HcclDataType dataType = HCCL_DATA_TYPE_RESERVED;
    const u64 inlineReduceMask = 1 << ATTR_POS_INLINE_REDUCE;
    ReduceType reduceType = (reduceAttr_ & inlineReduceMask) != 0 ? ReduceType::INLINE_REDUCE : ReduceType::TBE_REDUCE;
    u32 originalAlgTypeLevel1 = 0;
    u32 piplineSliceNum = 1;
    bool isDefaultPath = false;
    CopyPattern copyPattern = CopyPattern::ZCOPY;

    auto opMeta = HcclOpMetaInfo::GetOneForAllReduce(originalAlgTypeLevel1, dataType, reduceType, false,
        piplineSliceNum, hugeData, copyPattern, sliceNum, false, isDefaultPath);
    CHK_RET(InitTask(dispatcher_, stream_, opMeta.isEnableCache, opMeta.GetCacheKey()));

    HCCL_DEBUG("AHCExecutorBase phase[%u] end", fftsPhase_);

    fftsPhase_ = fftsPhase_ + 1;//更新Phase状态为下一个
    return HCCL_SUCCESS;
}


HcclResult AHCExecutorBase::RunInstance(const u32 rank, const std::vector<LINK> &links, const std::vector<Slice> &slices,
        std::unique_ptr<AlgTemplateBase> &tempAlg, HcclCMDType opType)
{
    HcclResult ret = HCCL_SUCCESS;

    bool  needRestFfts = false;

    // 判断是否关闭reducescatter的barrier
    if (!barrierSwitchOn_) {
        tempAlg->CloseBarrier();
    }
    // 调用算法执行
    if(opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
        needRestFfts = commAHCBaseInfo_->IsIntraAlgNB();
        ret = tempAlg->Prepare(inputMem_, inputMem_, outputMem_, count_, dataType_,
            stream_, reductionOp_, root_, slices, baseOffset_);
    } else if(opType == HcclCMDType::HCCL_CMD_ALLREDUCE) {
        needRestFfts = commAHCBaseInfo_->IsInterAlgNB();
        ret = tempAlg->Prepare(inputMem_, outputMem_, outputMem_, count_, dataType_,
            stream_, reductionOp_, root_, slices, baseOffset_);
    } else if(opType == HcclCMDType::HCCL_CMD_ALLGATHER) {
        needRestFfts = commAHCBaseInfo_->IsIntraAlgNB();
        ret = tempAlg->Prepare(outputMem_, outputMem_, outputMem_, count_, dataType_,
            stream_, reductionOp_, root_, slices, baseOffset_);
    }
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[AHCExecutorBase][RunInstance]rank[%u] prepare optype[%u] failed", rank, opType), ret);

    if (needRestFfts) {
        //子图phase重置
        CHK_RET(FftsRsetPhase(slices));
    }

    ret = tempAlg->RegisterProfiler(
        profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[AHCExecutorBase][RunInstance]rank[%u] registerProfiler optype[%u] failed", rank, opType), ret);
    
    ret = tempAlg->RunAsync(rank, links.size(), links);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[AHCExecutorBase][RunInstance]rank[%u] run optype[%u] failed", rank, opType), ret);

    return ret;
}

}   // ~~ namespace hccl
